package org.lisy.java.thread;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * ThreadPoolExecutor 线程池操作
 * 
 * @author lisy
 */
public class ThreadPoolExecutorOperate {

	public static void main(String[] args) {
		ThreadPoolExecutor executor = ThreadPoolExecutorUtil.getThreadPool();
		for (int i = 0; i < 20; i++) {
			MyRunnable myRunnable = new MyRunnable("task job-" + (i + 1));
			executor.execute(myRunnable);
		}
		
		/**
		 * shutdown: 线程池拒接收新提交的任务，同时等待线程池里的任务执行完毕后关闭线程池，要确保任务里不会有永久阻塞等待的逻辑
		 * shutdownNow: 线程池拒接收新提交的任务，同时立马关闭线程池，线程池里的任务不再执行
		 */
		executor.shutdown();
	}
	
}

/**
 * 创建线程池操作
 */
class ThreadPoolExecutorUtil {
	
	/**
	 * ThreadPoolExecutor 执行逻辑: 
	 * 1.判断核心线程数是否已满，核心线程数大小和 corePoolSize 参数有关，未满则创建线程执行任务
	 * 2.若核心线程池已满，判断队列是否满，队列是否满和 workQueue 参数有关，若未满则加入队列中
	 * 3.若队列已满，判断线程池是否已满，线程池是否已满和 maximumPoolSize 参数有关，若未满创建线程执行任务
	 * 4.若线程池已满，则采用拒绝策略处理无法执执行的任务，拒绝策略和 handler 参数有关
	 */
	private static ThreadPoolExecutor executor = null;
	
	public static synchronized ThreadPoolExecutor getThreadPool() {
		if (executor == null) {
			/**
			 * ThreadPoolExecutor 参数: 
			 * 线程池核心线程数量 2
			 * 线程池最大数量 4
			 * 空闲线程存活时间 60
			 * 时间单位 s
			 * 线程池所使用的缓冲队列 长度10
			 *  - 若使用 new SynchronousQueue<Runnable>() 创建的是长度为0的队列
			 *  - 若使用 LinkedBlockingQueue 需要指定长度，默认是长度为 Integer.MAX_VALUE，相当于无界，不会创建非核心线程，容易导致 OOM 异常
			 *  - 推荐使用 ArrayBlockingQueue 有界队列
			 * 线程池创建线程使用的工厂
			 * 线程池对拒绝任务的处理策略
			 *  - AbortPolicy: 默认策略。线程池队列满了丢掉这个任务并且抛出RejectedExecutionException异常
			 *  - DiscardPolicy: 线程池队列满了，会直接丢掉这个任务并且不会有任何异常
			 *  - DiscardOldestPolicy: 队列满了，会将最早进入队列的任务删掉腾出空间，再尝试加入队列
			 *  - CallerRunsPolicy: 如果添加到线程池失败，那么主线程会自己去执行该任务，不会等待线程池中的线程去执行
			 *  - 自定义: 实现RejectedExecutionHandler接口，并且实现rejectedExecution方法
			 */
			executor = new ThreadPoolExecutor(2, 5, 
					60L, TimeUnit.SECONDS,
					new LinkedBlockingQueue<>(10), 
					new MyThreadPoolFactory("pool-executor", false), 
					new MyRejectPolicy());
		}
		return executor;
	}
}

/**
 * 线程工厂
 */
class MyThreadPoolFactory implements ThreadFactory {

	// 线程计数
    private AtomicInteger threadIdx = new AtomicInteger(0);

    // 线程名前缀
    private String threadNamePrefix;
    
    // 是否守护线程
    private boolean threadDaemon;
    
    // 线程组
    private ThreadGroup threadGroup;

    public MyThreadPoolFactory(String prefix, boolean daemon) {
    	this.threadNamePrefix = prefix + "-thread-";
    	this.threadDaemon = daemon;
    	this.threadGroup = Thread.currentThread().getThreadGroup();
    }

    @Override
    public Thread newThread(Runnable r) {
        String name = threadNamePrefix + threadIdx.getAndIncrement();
        // stackSize：当前线程的指定栈大小，如果线程的创建者不指定大小，那默认值就是0，尽量不要修改这个值
        Thread thread = new Thread(threadGroup, r, name, 0);
        thread.setDaemon(threadDaemon);
        return thread;
    }

    public ThreadGroup getThreadGroup() {
        return threadGroup;
    }
}

/**
 * 自定义拒绝策略
 */
class MyRejectPolicy implements RejectedExecutionHandler{

	@Override
	public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
		// 判断是否为自定义 Runnable
        if (r instanceof MyRunnable) {
        	MyRunnable myRunnable = (MyRunnable) r;
        	// 打印处理被拒绝的任务内容，如果有回调函数也可以回调
            System.out.println("refuse " + myRunnable.getMessage());
        }
	}
    
}

/**
 * 自定义Runnable
 */
class MyRunnable implements Runnable {

	private String message;
	
	public MyRunnable(String msg){
		this.message = msg;
	}
	
	@Override
	public void run() {
		for (int i = 0; i < 10; i++) {
			System.out.println(message + ": " + i);
			try {
				TimeUnit.SECONDS.sleep(1);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public String getMessage() {
		return message;
	}
	
}
